package cn.doitedu.marketing.engine.functions;

import cn.doitedu.marketing.engine.beans.*;
import cn.doitedu.marketing.engine.buffer.BufferManager;
import cn.doitedu.marketing.engine.gauges.AvgGauge;
import cn.doitedu.marketing.engine.gauges.CountGauge;
import cn.doitedu.marketing.engine.gauges.QueryGauges;
import cn.doitedu.marketing.engine.service.TriggerModelConditionQueryService;
import cn.doitedu.marketing.engine.utils.ConnectionUtils;
import cn.doitedu.marketing.engine.utils.EventUtil;
import cn.doitedu.marketing.engine.utils.RuleSimulator2;
import cn.doitedu.marketing.engine.utils.StateDescContainer;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.kafka.common.metrics.stats.Avg;
import redis.clients.jedis.Jedis;

import java.util.List;
import java.util.Map;
import java.util.regex.MatchResult;

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-07-23
 * @desc 规则匹配计算核心函数
 * <p>
 * 规则：
 * 触发事件：  K事件，事件属性（ p2=v1 ）
 * 画像属性条件: tag87=v2, tag26=v1
 * 行为次数条件： 2021-06-18 ~ 当前 , 事件 C [p6=v8,p12=v5] 做过 >= 2次
 */
@Slf4j
public class RuleMatchKeyedProcessFunction extends KeyedProcessFunction<String, EventBean, RuleMatchResult> {
    ListState<EventBean> beansState;
    TriggerModelConditionQueryService triggerModelConditionQueryService;
    MarketingRule rule;
    QueryGauges queryGauges;

    @Override
    public void open(Configuration parameters) throws Exception {
        Connection hbaseConn = ConnectionUtils.getHbaseConnection();
        java.sql.Connection ckConn = ConnectionUtils.getClickhouseConnection();

        beansState = getRuntimeContext().getListState(StateDescContainer.getEventBeansDesc());

        // 获取规则
        rule = RuleSimulator2.getRule();
        log.debug("获取到规则, rule详情 :{}", rule);

        // 度量
        CountGauge ckQueryCountGauge = getRuntimeContext().getMetricGroup().gauge("ck-query-cnt", new CountGauge());
        AvgGauge ckQueryAvgMsGauge = getRuntimeContext().getMetricGroup().gauge("ck-avg-rate", new AvgGauge());
        CountGauge hbaseQueryCountGauge = getRuntimeContext().getMetricGroup().gauge("hbase-query-cnt", new CountGauge());
        AvgGauge hbaseQueryAvgMsGauge = getRuntimeContext().getMetricGroup().gauge("hbase-avg-rate", new AvgGauge());
        CountGauge bufferPutCountGauge = getRuntimeContext().getMetricGroup().gauge("buffer-put-cnt", new CountGauge());
        CountGauge bufferGetCountGauge = getRuntimeContext().getMetricGroup().gauge("buffer-get-cnt", new CountGauge());
        CountGauge bufferHitCountGauge = getRuntimeContext().getMetricGroup().gauge("buffer-hit-cnt", new CountGauge());
        queryGauges = new QueryGauges(hbaseQueryAvgMsGauge, ckQueryAvgMsGauge, ckQueryCountGauge, hbaseQueryCountGauge, bufferPutCountGauge, bufferGetCountGauge, bufferHitCountGauge);

        Jedis jedis = ConnectionUtils.getRedisConnection();
        BufferManager bufferManager = new BufferManager(jedis, queryGauges);
        triggerModelConditionQueryService = new TriggerModelConditionQueryService(hbaseConn, ckConn, beansState, queryGauges, bufferManager);
    }

    @Override
    public void processElement(EventBean event, Context ctx, Collector<RuleMatchResult> out) throws Exception {

        // 将当前收到的event存入flink state中
        beansState.add(event);

        // 规则计算 TODO 判断当前用户是否已经到达推送次数上限
        if (!ruleMatch(event, rule)) return;

        // 判断规则是否要注册timer
        if (rule.isOnTimer()) {
            // timer规则，注册 time service
            ctx.timerService().registerEventTimeTimer(3000);
        } else {
            // 非timer规则，输出规则匹配结果
            RuleMatchResult ruleMatchResult = new RuleMatchResult(event.getDeviceId(), rule.getRuleId(), event.getTimeStamp(), System.currentTimeMillis());
            out.collect(ruleMatchResult);
        }

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<RuleMatchResult> out) throws Exception {
        // 遍历所有timer rule
        // for(MarketingRule rule: ruleList)
        List<EventCombinationCondition> eventCombinationConditionList = rule.getEventCombinationConditionList();
        for (EventCombinationCondition eventCombinationCondition : eventCombinationConditionList) {
            EventBean eventBean = new EventBean();
            eventBean.setTimeStamp(timestamp-3000);
            boolean b = triggerModelConditionQueryService.matchEventCombinationCondition(eventCombinationCondition, ctx.getCurrentKey(), eventBean);
            if(b) return;
            RuleMatchResult ruleMatchResult = new RuleMatchResult(ctx.getCurrentKey(), rule.getRuleId(), timestamp - 3000, timestamp);
            out.collect(ruleMatchResult);
        }

    }

    private boolean ruleMatch(EventBean event, MarketingRule rule) throws Exception {
        EventCondition triggerEvent = rule.getTriggerEvent();
        if (!EventUtil.eventMatchCondition(event, triggerEvent)) return false;
        log.debug("规则被触发，触发事件为, eventId:{},deviceId:{},props:{}", event.getEventId(), event.getDeviceId(), event.getProperties());


        Map<String, String> userProfileConditions = rule.getUserProfileConditions();
        if (userProfileConditions != null && userProfileConditions.size() > 0) {
            boolean b = triggerModelConditionQueryService.matchProfileCondition(event.getDeviceId(), rule);
            if (!b) return false;
        }

        List<EventCombinationCondition> eventCombinationConditionList = rule.getEventCombinationConditionList();
        if (eventCombinationConditionList != null && eventCombinationConditionList.size() > 0) {
            for (EventCombinationCondition eventCombinationCondition : eventCombinationConditionList) {
                boolean b = triggerModelConditionQueryService.matchEventCombinationCondition(eventCombinationCondition, event.getDeviceId(), event);
                if (!b) return false;
            }
        }
        return true;
    }


}
